Система асинхронных задач

Разбор документов, построение индексов, генерацию резюме и вопросов, извлечение графа, генерацию Wiki, синхронизацию источников данных и массовые операции планирует система задач. Стандартное развёртывание использует asynq и Redis; режим Lite использует исполнитель без Redis. Оба режима разделяют логику обработки задач, но различаются способом выполнения и возможностями эксплуатации.

Общая архитектура: два режима выполнения

У платформы есть два режима выполнения задач, выбираемых по форме развёртывания:

• Режим asynq (стандартное развёртывание): задачи через asynq.Client сериализуются в JSON payload и записываются в очередь Redis, откуда их потребляют несколько независимых asynq.Server (worker pool). RunAsynqServer() из internal/router/task.go строит единый asynq.ServeMux и запускает его на 6 pool.
• Режим Lite (одна машина / приложение macOS, без Redis): SyncTaskExecutor из internal/router/sync_task.go реализует тот же интерфейс interfaces.TaskEnqueuer; Enqueue сразу передаёт задачу в goroutine на выполнение, поддерживает опции ProcessIn / ProcessAt (задержка), MaxRetry, Timeout и Deadline, задаваемые как в asynq.NewTask, так и в Enqueue (последний имеет приоритет). Задачи без заданного таймаута не ограничены по времени — это отличается от стандартных 30 минут в asynq. Повтор выполняется с линейным откатом (attempt * 5s, максимум 30s), а задачи, вернувшие asynq.SkipRetry, больше не повторяются. panic внутри handler перехватывается и трактуется как сбой, не приводя к выходу процесса. После исчерпания повторов выполняется то же завершение, что и у колбэка недоставленных задач стандартного режима: задачи типа «документ» помечают знание как failed.

// internal/router/sync_task.go
// SyncTaskExecutor executes tasks synchronously (in a goroutine) without Redis.
// Used in Lite mode as a drop-in replacement for *asynq.Client.

Наборы зарегистрированных handler в обоих режимах полностью совпадают (сравните RunAsynqServer и RegisterSyncHandlers), что гарантирует, что семантика задач не «дрейфует» в зависимости от формы развёртывания.

Роль Redis в системе

Роль	Описание	Расположение исходного кода
asynq broker	Все очереди задач (pending list, scheduled/retry ZSET, archived ZSET) хранятся в Redis; dequeue атомарен (BRPOPLPUSH), что гарантирует выполнение одной задачи только одним worker	internal/router/task.go getAsynqRedisClientOpt()
Источник данных для инспекции задач	asynq.Inspector + прямое постраничное чтение LPos/ZRank/ZRevRank	internal/router/task_inspector.go
Мьютекс Wiki ingest	wiki:active:<kbID>, блокировка finalize, блокировка slug — все SetNX + TTL	internal/application/service/wiki_ingest.go, wiki_ingest_batch.go
Счётчик подзадач мультимодальности	Счётчик завершённых подзадач изображений (DECR), последняя attempt запускает finalize; в режиме Lite используется внутрипроцессный счётчик	Сервисы, связанные с image_multimodal
Ограничение частоты	ZSET ограничения частоты со скользящим окном (см. документацию по наблюдаемости)	internal/ratelimit/limiter.go

Параметры подключения к Redis берутся из переменных окружения REDIS_ADDR / REDIS_USERNAME / REDIS_PASSWORD / REDIS_DB / конфигурации TLS. Таймауты чтения и записи управляются WEKNORA_REDIS_OP_TIMEOUT_MS, по умолчанию 500ms (таймаут записи вдвое больше, чтобы гасить блокировку головы очереди):

// internal/router/task.go
const defaultRedisOpTimeoutMs = 500
opt := &asynq.RedisClientOpt{
    Addr:        os.Getenv("REDIS_ADDR"),
    ReadTimeout: time.Duration(timeoutMs) * time.Millisecond,
    WriteTimeout: time.Duration(timeoutMs*2) * time.Millisecond,
    ...
}

Список типов задач

Константы типов задач определены в internal/types/task.go:

Тип задачи	Константа	Назначение	Очередь
document:process	TypeDocumentProcess	Точка входа разбора документа (DocReader / сегментация / векторизация)	default
manual:process	TypeManualProcess	Обновление знания вручную (cleanup + повторная индексация)	default
temporary_document:process	TypeTemporaryDocumentProcess	Разбор временного документа сессии (вложения чата)	chat_attachment
knowledge:post_process	TypeKnowledgePostProcess	Единое планирование постобработки знания (fan-out подзадач обогащения)	postprocess
knowledge:auto_tag	TypeKnowledgeAutoTag	Автоматическая привязка уже имеющихся у документа тегов	summary
memory:extract	TypeMemoryExtract	Фоновое извлечение личной памяти	memory
summary:generation	TypeSummaryGeneration	Генерация резюме + профиля документа	summary
kb:profile	TypeKnowledgeBaseProfile	Описание базы знаний (агрегация профиля + один вызов небольшой модели, пропускается при неизменном хэше)	summary
datatable:summary	TypeDataTableSummary	Резюме таблицы	summary
image:multimodal	TypeImageMultimodal	OCR изображений + VLM Caption	multimodal
chunk:extract	TypeChunkExtract	Извлечение сущностей/связей графа (по chunk)	graph
question:generation	TypeQuestionGeneration	Генерация вопросов (fan-out по пакетам chunk)	question
datasource:sync	TypeDataSourceSync	Синхронизация источника данных	sync
faq:import	TypeFAQImport	Импорт FAQ (включая dry run)	low (maintenance)
kb:clone	TypeKBClone	Копирование базы знаний	low
kb:delete	TypeKBDelete	Удаление базы знаний	low
index:delete	TypeIndexDelete	Удаление индекса	low
knowledge:list_delete	TypeKnowledgeListDelete	Массовое удаление знаний	low
knowledge:list_reparse	TypeKnowledgeListReparse	Массовый повторный разбор	low
knowledge:move	TypeKnowledgeMove	Перемещение знания	low
wiki:ingest	TypeWikiIngest	Генерация/синхронизация страниц Wiki	wiki
wiki:finalize	TypeWikiFinalize	Завершение уровня KB для Wiki (антидребезг: перестроение индекса/очистка битых ссылок/перекрёстные ссылки)	wiki

Все структуры payload (например, DocumentProcessPayload, ImageMultimodalPayload) встраивают types.TracingContext для передачи Langfuse/W3C traceparent между процессами (см. документацию по наблюдаемости), а также единообразно несут поля маршрутизации tenant_id / knowledge_id / knowledge_base_id и т. п., используемые для архивации недоставленных задач и сопоставления при отмене.

Автоматические теги, резюме и описания баз знаний используют общую очередь summary и относятся к необязательным задачам обогащения; автоматические теги ставятся в очередь только когда у базы знаний документов включён auto_tag_config, а описание базы знаний (kb:profile) запускается из финального состояния/при удалении/перемещении только при включённом profile_config и дедуплицируется по 30-секундному окну; их сбой не влияет на уже завершённый разбор. Извлечение памяти использует отдельную очередь memory, потребляется enrichment pool и участвует в эластичном заимствовании shared pool с весом 1, при этом общее число worker pool остаётся равным 6.

Задачи памяти дедуплицируются по субъекту-человеку и агрегируются с задержкой; extract_cursor, pending_sessions и время планирования в memory_subjects используются для продолжения, чтобы не запускать извлечение моделью сразу при каждом вопросе. При отключённой в пространстве памяти или при write_mode, отличном от auto, фоновая дистилляция не запускается. Синхронный исполнитель Lite также регистрирует задачи автоматических тегов и памяти, следуя тем же бизнес-переключателям.

Топология Worker Pool и стратегия управления

queueDefinitions в internal/types/task.go — единственный источник истины (single source of truth) топологии очередей; построение worker server (QueueWeightsForPool) и отображение в панели эксплуатации (QueueStats) используют этот реестр совместно, что предотвращает расхождение весов.

Шесть независимых worker pool

Каждый pool — это независимый asynq.Server с жёстко изолированным параллелизмом (а не с весовым предпочтением). Параллелизм по умолчанию и ключи конфигурации (ключ system_settings / переменная окружения, см. types.ResolveWorkerPoolConcurrency):

Pool	Параллелизм по умолчанию	Потребляемые очереди (вес)	Ключ конфигурации / переменная окружения
core	8	default(1), chat_attachment(3)	asynq.core_concurrency / WEKNORA_ASYNQ_CORE_CONCURRENCY
postprocess	2	postprocess(1)	asynq.postprocess_concurrency / WEKNORA_ASYNQ_POSTPROCESS_CONCURRENCY
enrichment	12	summary(2), multimodal(1), graph(1), question(1), memory(1)	asynq.enrichment_concurrency / WEKNORA_ASYNQ_ENRICHMENT_CONCURRENCY
maintenance	4	sync(2), low(1)	asynq.maintenance_concurrency / WEKNORA_ASYNQ_MAINTENANCE_CONCURRENCY
shared (эластичный слой)	6	очереди core + enrichment с SharedWeight > 0	asynq.shared_concurrency / WEKNORA_ASYNQ_SHARED_CONCURRENCY
wiki	8	wiki(1)	asynq.wiki_concurrency / WEKNORA_WIKI_ASYNQ_CONCURRENCY

Ключевые решения (подтверждаются комментариями в исходном коде):

• Гарантированная ёмкость + эластичное заимствование: core/postprocess/enrichment/maintenance обеспечивают минимальную гарантированную ёмкость; shared pool подписан одновременно на очереди core и enrichment, поэтому простаивающая ёмкость может быть заимствована любой стадией (NewSharedAsynqServer: dequeue в Redis атомарен, и при подписке нескольких server на одну очередь каждая задача всё равно выполняется один раз). post-process и maintenance не участвуют в shared pool: первому нужна отдельная гарантия задержки, второй выполняется долго и мог бы поглощать пиковую ёмкость интерактивных задач. Область определяется QueueWeightsForSharedPool.
• Жёсткая изоляция Wiki: wiki pool тянет только очередь wiki, чтобы конвейер разбора и генерация Wiki не «морили голодом» друг друга (комментарий NewWikiAsynqServer).
• Приоритет вложений чата: в core pool вес chat_attachment равен 3, что выше 1 у default, поэтому массовый импорт KB не ставит в очередь интерактивные загрузки чата.
• Совместимость при скользящем обновлении: физическое имя очереди Redis для константы QueueMaintenance остаётся прежним "low", поэтому задачи, поставленные старой версией, всё ещё могут быть обработаны во время скользящего развёртывания.

Оценка ёмкости и масштабирование

Старая агрегированная конфигурация asynq.concurrency / WEKNORA_ASYNQ_CONCURRENCY выведена из использования, и развёртываниям следует перейти на конфигурацию по пулам из таблицы выше. После изменения настроек требуется перезапуск сервиса. Первые пять пулов в сумме дают 32 worker на экземпляр по умолчанию, 8 worker Wiki считаются отдельно.

Приведённую ниже оценку можно использовать как отправную точку, а затем уточнять её по панели времени выполнения и реальной нагрузке:

Требуемые worker ≈ ceil(пиковая скорость поступления задач × среднее время выполнения / 0.70)

Здесь 0.70 — это пример целевой утилизации, а не конфигурация системы или фиксированная гарантия ёмкости. Скорость поступления нужно считать по числу задач после fan-out: один документ может породить несколько пакетов вопросов, задачи графа по каждому chunk и несколько задач изображений. Само число очередей не отражает пропускную способность.

Worker контролирует, сколько задач разрешено выполнять одновременно на каждом экземпляре сервиса; квоты модели контролируют параллелизм, RPM и TPM между репликами; у DocReader, векторной базы, базы данных и объектного хранилища есть собственные потолки ёмкости. Когда ожидание из-за ограничения модели уже велико, увеличение числа worker лишь прибавляет ожидающих. Следует опираться на время ожидания самой старой задачи, суммарную ёмкость активных экземпляров, утилизацию worker и состояние нижележащих ресурсов: увеличивать соответствующий pool, только когда у нижележащих ресурсов есть запас и накопление продолжает расти; при заполненном DocReader снижать приём в core.

Схема архитектуры Worker Pool

Управление middleware

RunAsynqServer (internal/router/task.go) последовательно устанавливает на одном mux четыре middleware:

1. asynqdl.MiddlewareWithCallback (недоставленные задачи) — должен устанавливаться первым, чтобы видеть исходную ошибку, возвращаемую handler (последующие middleware могут её преобразовывать). См. Повторы при сбое и обработка недоставленных задач.
2. asynqdl.RecoverMiddleware — превращает panic в handler в обычную ошибку задачи. Сам asynq восстанавливает panic только вне всех middleware, и без него колбэк недоставленных задач не увидит ошибку, а документ последней попытки навсегда останется в состоянии processing.
3. backgroundTaskMiddleware — помечает context каждой задачи через types.WithBackgroundTask, заставляя регулятор параллелизма чата для каждой модели (chat concurrency governor) ограничивать LLM-вызовы при ingestion/enrichment, но не влияя на интерактивный чат пользователей.
4. langfuse.AsynqMiddleware — при выключенном Langfuse работает как сквозной; при включённом продолжает вышестоящий HTTP-trace либо открывает независимый trace, оборачивая выполнение handler в SPAN.

Стратегия отката повторных попыток

По умолчанию используется экспоненциальный откат asynq (примерно 10s, 40s, 90s, 2.5m…), но для конфликта блокировки Wiki ingest сделана кастомизация (asynqRetryDelayFunc):

// internal/router/task.go
func asynqRetryDelayFunc(n int, e error, t *asynq.Task) time.Duration {
    if errors.Is(e, service.ErrWikiIngestConcurrent) {
        return wikiIngestRetryDelay // фиксированные 15s
    }
    return asynq.DefaultRetryDelayFunc(n, e, t)
}

Причина: TTL осиротевшей блокировки ≤ 60s, поэтому повтор через фиксированные 15s почти наверняка успешен; экспоненциальный откат, наоборот, заставил бы KB после аварийного перезапуска застрять на 7–10 минут.

Конечный автомат жизненного цикла задачи

Состояния времени выполнения на стороне asynq (runtimeTaskState в internal/router/task_inspector.go отображается в types.RuntimeTaskState): pending, active, scheduled, retry, archived, completed. Бизнес-состояние строки знания parse_status (internal/types/knowledge.go): pending → processing → finalizing → completed, а также failed / deleting / cancelled.

Соответствующие состояния строки знания (управляемые задачей):

Инспекция задач, отмена и панель эксплуатации (TaskInspector)

internal/router/task_inspector.go реализует interfaces.TaskInspector; в режиме asynq он опирается на asynq.Inspector + нативный Redis client; в режиме Lite используется noopTaskInspector (goroutine нельзя изъять до запуска, поэтому единственный сигнал остановки — отмена через контрольные точки).

Отмена по знанию / базе знаний

• CancelTasksForKnowledge(ctx, knowledgeID): сканирует четыре состояния pending/scheduled/retry/active во всех зарегистрированных очередях (queuesScanned берётся из types.QueueDefinitions()), обрабатывая совпадение по knowledge_id в payload. Белый список отменяемых типов задач taskTypesForKnowledgeCancel: document:process, manual:process, image:multimodal, knowledge:post_process, question:generation, summary:generation, chunk:extract (без импорта FAQ и задач уровня базы знаний).
• Процесс отмены состоит из трёх этапов (cancelMatchingTasks): ① сначала чисто удаляются задачи из очереди; ② после снимка активных задач вызывается Inspector.CancelProcessing для отправки сигнала, и в течение 1s окна стабилизации выполняется опрос (с интервалом 25ms) для удаления записей, перешедших в retry из-за context.Canceled (deleteCancelledTransitions); ③ задачи в очереди сканируются ещё раз, чтобы поймать нижележащие задачи, поставленные в очередь во время отмены.
• CancelTasksForKnowledgeBase: очистка осиротевших задач после удаления KB; kb:delete и index:delete явно исключены (они несут снимок и отвечают за фактическую очистку хранилища — их удаление привело бы к утечке ресурсов). В сопоставлении также участвуют семантические поля KB операций clone/move (source_id/target_id/source_kb_id/target_kb_id).
• Всё выполняется best-effort: при сбоях Redis пишется предупреждение в лог и ошибка проглатывается, а API отмены всё равно возвращает успех.
• HasQueuedTasksForKnowledge: только для чтения — очистка housekeeping использует его, чтобы отличить «в очереди, но не осиротевшие» строки и не пометить их ошибочно как failed. HasQueuedDeleteTasksForKnowledge специально сопоставляет массовый payload knowledge:list_delete и служит для восстановления застрявшего удаления.
• QueuedKnowledgeIDs: за одно сканирование всех очередей возвращает множество ID знаний, на которые всё ещё ссылаются задачи в очереди; по нему интерфейс списка знаний отдаёт stall_state (queued / stalled), результат кэшируется на 60 секунд и разделяется между конкурентными запросами, а при сбое сканирования не кэшируется.

Панель эксплуатации (SystemAdmin Runtime Dashboard)

• QueueStats(): для каждой очереди вызывается GetQueueInfo и выводится types.QueueStat (size/pending/active/scheduled/retry/archived/completed, processed/failed за сутки, paused, latency_ms (возраст самой старой pending-задачи), потребление памяти), а также статические метаданные pool/weight. Для никогда не создававшихся очередей возвращается строка с нулями (isAsynqQueueNotFound также совместим с внутренней строкой ошибки NOT_FOUND, просочившуюся в asynq v0.26, см. task_inspector_errors.go).
• ListRuntimeTasks(): прямое постраничное чтение на основе ключей Redis asynq:{<queue>}:<state> — pending/active это LIST (новые впереди), scheduled/retry это ZSET по возрастанию NextProcessAt, archived/completed по убыванию балла. Курсор — это закодированное в base64 окно якорей (не более 32 якорей, runtimeTaskCursorMaxAnchors); при исчезновении якоря (задача завершена/повторена/удалена) разбиение на страницы можно продолжать. В payload проецируются только белые списки метаданных маршрутизации (ID tenant/kb/knowledge/task/sync и т. п.), никогда не раскрывается содержимое документов или секреты.
• Действия над задачами ограничиваются проверкой состояния runtimeTaskActions: cancel (pending/active/scheduled/retry и отменяемый тип), run_now (scheduled/retry/archived, asynq сохраняет счётчик повторов), delete (только archived); кроме того, PurgeArchivedRuntimeTasks одной командой очищает набор archived отдельной очереди.
• WorkerServerStats(): читает heartbeat серверов asynq (параллелизм, число активных worker, статус, веса очередей) и после агрегации между репликами различает «настроенную ёмкость одного экземпляра» и «фактическую ёмкость кластера».

Соответствующий HTTP API (internal/router/routes_auth_tenant.go, с гейтингом SystemAdmin + capability платформенного API Key):

Метод	Путь	Описание
GET	/api/v1/system/admin/runtime/queues	Снимок глубины очередей + heartbeat worker
GET	/api/v1/system/admin/runtime/queues/:queue/tasks	Постраничный список задач по состоянию с курсором
POST	/api/v1/system/admin/runtime/queues/:queue/tasks/:task_id/actions/:action	cancel / run_now / delete (с записью в аудит платформы)
DELETE	/api/v1/system/admin/runtime/queues/:queue/archived	Очистка archived (с записью в аудит платформы system.queue_archived_purged)

Повторы при сбое и обработка недоставленных задач

Middleware недоставленных задач asynq (internal/middleware/asynqdl/asynqdl.go)

• Строка task_dead_letters пишется только при сбое последней попытки (isFinalAttempt: retried >= max_retry), чтобы мгновенный сбой не порождал строку каждый раз.
• buildDeadLetter с помощью терпимого payloadProbe извлекает из произвольного payload tenant_id / knowledge_base_id / kb_id / knowledge_id / source_kb_id, а inferScope по «радиусу поражения» выводит scope (knowledge_base > knowledge > tenant > unknown). Payload сохраняется как есть (может пригодиться для будущего воспроизведения), last_error обрезается до 8KB.
• Вставка выполняется best-effort: сбой БД только логируется, а исходная ошибка задачи всегда пробрасывается в asynq как есть (попадая в archived).
• Колбэк OnDeadLetter (newDeadLetterKnowledgeFailer из internal/router/task.go): когда у document:process / knowledge:post_process / manual:process исчерпаны повторы, и только если строка знания всё ещё в pending / processing / finalizing, а tenant, база знаний и payload согласованы, одним условным обновлением записываются сразу parse_status=failed + error_message (это исключает частичное обновление и не перезаписывает уже отменённые или удалённые строки; задачи, завершившиеся через SkipRetry, колбэк не запускают), а также вызывается SpanTracker.FinalizeAttempt, закрывающий корневой span соответствующей attempt, чтобы на временной шкале больше не отображалось «выполняется». Для knowledge:list_delete есть отдельная ветка markKnowledgeListDeleteFailed. image:multimodal не помечает родительское знание как failed (finalize-on-last-attempt уже гарантирует прогресс). Колбэк выполняется с context.Background(), а panic перехватывается, и исходная ошибка задачи никогда не изменяется.

Персистентная очередь задач и недоставленные задачи уровня сервиса (internal/application/repository/task_queue.go)

Таблица task_pending_ops — персистентная замена очереди-списка Redis (не теряется при перезапуске, без вытеснения по TTL); идентичность очереди — это тройка (task_type, scope, scope_id), и в настоящее время основной её потребитель — Wiki ingest:

• Enqueue / EnqueueIfKnowledgeBaseActive: последний в транзакции с блокировкой строки Postgres SHARE проверяет, что KB ещё жива и что её tenant не удалён мягко, не позволяя записать новую персистентную работу после удаления KB или tenant.
• Удалённые tenant: удаление tenant выполняется только как мягкое, его базы знаний и строки task_pending_ops остаются. При восстановлении после запуска сервиса очищаются необработанные строки мягко удалённых tenant; задачи Wiki ingest / finalize перед вызовом модели проверяют, жив ли tenant, и при удалённом tenant отбрасывают очередь этой базы знаний, больше не порождая запросов к модели.
• ClaimBatch: атомарное «всей группой» завладение по dedup_key (=документ). Ключевой инвариант: несколько op одного документа (например, ingest, за которым следует retract) никогда не разводятся по двум конкурентным пакетам; ключ со свежим claim (claimed_at >= staleBefore) целиком пропускается, а позднее пришедший родственный op ждёт завершения владельца или истечения claim. В Postgres через строку-якорь каждого ключа FOR UPDATE SKIP LOCKED гарантируется, что конкурентные «забирающие» получают непересекающиеся наборы ключей; SQLite (Lite/тесты) полагается на движок с единственным писателем. Порядок завладения — по возрастанию fail_count, а при равном числе сбоев по порядку постановки в очередь: документы, падающие снова и снова, не стоят вечно в начале очереди, а новые документы не голодают (в режиме Lite PeekBatch использует тот же порядок).
• IncrFailCount (атомарное увеличение одним round-trip через UPDATE ... RETURNING) вместе с серверным лимитом (wikiMaxFailRetries у wiki): после превышения лимита op переносится из task_pending_ops в task_dead_letters (internal/application/service/wiki_ingest.go напрямую вызывает deadLetterRepo.Insert).
• ReleaseByIDs / DeleteByIDs / DeleteByScope / DeleteByDedupKey / PendingCount обеспечивают освобождение, подтверждение потребления, очистку при жизненном цикле KB и наблюдение за накоплением.

Репозиторий недоставленных задач taskDeadLetterRepository предоставляет ListByScope / ListByTaskType (постранично по курсору в обратном порядке id, limit 1–200) и DeleteByID; для эксплуатации можно напрямую запрашивать сбои SQL по типу задачи / scope / tenant, без перелистывания логов.

Резервный механизм: очистка housekeeping

internal/application/service/knowledge_housekeeping.go: cron запускает раунд каждые 5 минут (0 */5 * * * *); это последняя линия обороны помимо повторов asynq, колбэка недоставленных задач и finalize мультимодальности (сценарии, когда worker убит посреди handler, defer не выполнился и т. п.). Можно отключить через WEKNORA_HOUSEKEEPING_ENABLED=false.

Порог — max(1h, WEKNORA_DOCUMENT_PROCESS_TIMEOUT) + 10min. Каждый раунд включает:

Очистка	Объект	Действие
Зависший разбор	pending / processing / finalizing дольше порога, heartbeat span тоже дольше порога, и ни в очереди asynq, ни в персистентной очереди Wiki нет связанных задач	Ставится failed, а error_message указывает, на какой стадии остановилось и когда был последний прогресс; незавершённые span последней attempt закрываются как TASK_STALLED (застрявшая стадия — failed, остальные — cancelled)
Зависшее резюме	summary_status = processing дольше 1 часа	summary_status ставится в failed
Зависшее удаление	deleting дольше порога, и в очереди нет перекрывающей его задачи удаления	Ставится failed с указанием причины, документ снова виден и его можно удалить повторно; при сбое проверки очереди действие откладывается
Повторный запуск очереди Wiki	Документы, застрявшие в finalizing только из-за непотреблённой персистентной очереди Wiki	Для соответствующей базы знаний заново ставится в очередь задача-триггер Wiki, не более одного раза на базу знаний за период порога

Постоянные логи «tasks still queued (backpressure, not stuck)» означают, что узкое место — ёмкость очередей, и следует увеличить параллелизм соответствующего worker pool, а не подозревать ошибку инспекции.

Шина событий (internal/event)

Шина событий служит для внутрипроцессного распределения потоковых событий сессий/агента (например SSE-отправка, колбэки IM) и дополняет asynq (межпроцессные персистентные задачи).

Структура и гарантии доставки

// internal/event/event.go
type Event struct {
    ID        string                 // ID события (автоматически генерируемый UUID для отслеживания потоковых обновлений)
    Type      EventType
    SessionID string
    Data      interface{}
    Metadata  map[string]interface{}
    RequestID string
}

• EventBus.On(type, handler) регистрирует (для одного типа может быть несколько handler), Off / Clear удаляют; HasHandlers / GetHandlerCount служат для запросов.
• Синхронный режим (NewEventBus, по умолчанию): Emit выполняет handler по порядку, и ошибка любого handler немедленно возвращается (at-most-once, ошибка прерывает последующие handler).
• Асинхронный режим (NewAsyncEventBus): Emit запускает для каждого handler goroutine, fire-and-forget, ошибки отбрасываются, panic перехватывается и логируется.
• EmitAndWait: в обоих режимах выполняет все handler параллельно и ждёт завершения, собирая ошибки и panic.
• Гарантия доставки — внутрипроцессная, неперсистентная: при отсутствии зарегистрированных handler событие молча отбрасывается (возвращается nil); при аварии процесса события в полёте теряются. Для персистентности следует использовать asynq или task_pending_ops.
• global.go предоставляет глобальный синглтон (event.On / event.Emit); на практике потоковая обработка уровня сессии использует отдельный экземпляр bus (см. подписчиков).
• middleware.go предоставляет middleware для handler: WithLogging (логи запуска/сбоя), WithTiming (запись времени в metadata), WithRecovery (преобразование panic в PanicError), комбинируемые через Chain / ApplyMiddleware.
• EventBusAdapter из adapter.go адаптирует *EventBus под types.EventBusInterface, избегая циклических зависимостей.

Список типов событий (internal/event/event.go)

Группа	Типы событий
Обработка запроса	query.received, query.validated, query.preprocess, query.rewrite, query.rewritten
Поиск	retrieval.start, retrieval.vector, retrieval.keyword, retrieval.entity, retrieval.complete
Переранжирование	rerank.start, rerank.complete
Объединение	merge.start, merge.complete
Генерация чата	chat.start, chat.complete, chat.stream
Жизненный цикл Agent	agent.query, agent.plan, agent.step, agent.tool, agent.complete
Потоковый режим Agent (обратная связь в реальном времени)	thought, tool_call, tool_result, reflection, references, final_answer
Ручное одобрение инструментов MCP	tool_approval_required, tool_approval_resolved
Авторизация MCP OAuth в рамках сессии	mcp_oauth_required, mcp_oauth_resolved
Ошибки / сессия / управление	error, session_title, stop

Структуры данных для каждого типа события определены в internal/event/event_data.go (например, AgentToolCallData несёт tool_call_id/tool_name/arguments/hint, а AgentFinalAnswerData — content/done/is_fallback и т. п.).

Основные подписчики

Подписчик	Исходный код	Подписка
Потоковый handler SSE Agent	internal/handler/session/agent_stream_handler.go	thought, tool_call, tool_result, references, final_answer, reflection, error, session_title, agent.complete, а также четыре типа одобрения инструментов и MCP OAuth
Handler вопрос-ответа по знаниям	internal/handler/session/qa.go, helpers.go	thought, final_answer, stop
Интеграция с IM (WeCom и др.)	internal/im/service.go	final_answer, error, references, agent.complete, thought, tool_call, tool_result, mcp_oauth_required и др., транслируемые в сообщения соответствующих IM-платформ

Пакет internal/runtime

Пакет небольшой и представляет собой инфраструктуру времени выполнения, а не логику worker:

• container.go: init() создаёт глобальный *dig.Container (uber dig), а GetContainer() служит для регистрации/разрешения зависимостей в пакетах. Через него собираются все asynq server, handler и repository (фактическая масштабная сборка выполняется в internal/container/container.go).
• server.go: MarkServerStarted() / ServerStartedAt() / ServerUptime() — фиксация момента запуска процесса для отображения uptime в панели эксплуатации.
• startup.go: SilenceGinRouteSpam() подавляет около 150 строк логов регистрации маршрутов Gin, собирая их в одну (LogGinRouteCount); LogStartupEnv() печатает баннер из отобранных переменных окружения (для чувствительных значений показывается только set (N chars)) и выдаёт явные предупреждения о типичных «граблях» (например, когда длина SYSTEM_AES_KEY не равна 32 и шифрование фактически отключено, или при REDIS_TLS_INSECURE_SKIP_VERIFY=true).

Как мониторить задачи

1. Панель эксплуатации / Runtime API (раздел 6.2): глубина очередей, задержка самой старой pending-задачи (latency_ms), processed/failed за сутки, heartbeat worker; просмотр задач по состоянию, last_error, retried/max_retry, выполнение run_now/cancel/delete.
2. SQL по таблице недоставленных задач: SELECT * FROM task_dead_letters WHERE scope='knowledge_base' AND scope_id='<kbID>' ORDER BY id DESC; либо агрегирование доли сбоев по task_type; PendingCount / enqueued_at из task_pending_ops позволяют обнаружить накопление, которое никогда не разгружалось.
3. Логи: на стороне worker всё идёт через internal/logger, ключевые префиксы — [TaskInspector] (отмена/инспекция), asynq dead-letter, [SyncTask] (режим Lite), [Housekeeping]; при запуске каждый pool печатает asynq <pool> server starting with concurrency=....
4. Langfuse trace: при включении каждая задача asynq — это SPAN asynq.<task_type> (с метаданными queue, retry, размера payload), в том же trace, что и запустивший её HTTP-запрос (см. документацию по наблюдаемости).
5. Аудит платформы: операции run_now/delete/purge над задачами archived пишутся в audit_logs (действия system.queue_task_*) и позволяют установить ответственного.

Справка по реализации

Модуль	Путь исходного кода
Регистрация задач и построение worker pool	internal/router/task.go
Синхронный исполнитель режима Lite (без Redis)	internal/router/sync_task.go
Инспекция задач / отмена / панель эксплуатации	internal/router/task_inspector.go, internal/router/task_inspector_errors.go
Топология очередей и определение типов задач	internal/types/task.go
Middleware недоставленных задач	internal/middleware/asynqdl/asynqdl.go
Персистентная очередь задач / репозиторий недоставленных задач	internal/application/repository/task_queue.go
Модели недоставленных задач / отложенных операций	internal/types/task_dead_letter.go, internal/types/task_pending_op.go
Шина событий	internal/event/ (event.go, event_data.go, global.go, middleware.go, adapter.go)
Вспомогательные средства времени выполнения (DI-контейнер, баннер запуска, uptime)	internal/runtime/ (container.go, server.go, startup.go)
Резервная очистка зависших задач	internal/application/service/knowledge_housekeeping.go