Быстрый приём: сохранение в catched, добавление в qBittorrent — шаг worker'а
Приём (Ingest) стал быстрым: синхронно только парс magnet, синтез контекста из полей ссылки, атомарный дедуп и запись загрузки в новое состояние `catched` — ответ клиенту сразу. Медленный вывод имени (LLM) и добавление в qBittorrent вынесены в асинхронный шаг машины состояний, который двигает worker. - store: состояние `catched` (нетерминальное, активная группа); атомарный переход PromoteCatched (catched → downloading + display_name) с гардом state='catched' (ре-валидация после сетевых вызовов вне блокировки) - ingest: убраны namer/qbt из пути приёма; пишем `catched`, отвечаем сразу - worker.processCatched: вне w.mu выводит имя и qbt.Add, под w.mu — короткий переход; сбой add оставляет catched (ретрай тиком); предохранитель catch_timeout → failed(qbit_add)+notify; catched исключён из проверок пропажи - config: worker.catch_timeout (дефолт 10m) - веб-UI: бейдж catched, активная группа, самозавершающийся htmx-поллинг карточки/страницы до перехода в downloading; Telegram-текст без сырого catched - OpenSpec: дельты ingest/download-tracking/web-ui влиты в спеки, change заархивирован Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
+29
-99
@@ -1,6 +1,9 @@
|
||||
// Package ingest — use-case приёма загрузки, общий для всех транспортов
|
||||
// (HTTP, Telegram, CLI). Принимает источник + контекст, отдаёт источник в
|
||||
// qBittorrent и заводит/находит задачу в БД.
|
||||
// Package ingest — use-case быстрого приёма загрузки, общий для всех
|
||||
// транспортов (HTTP, Telegram, CLI). Синхронно только парсит источник,
|
||||
// синтезирует контекст из полей ссылки, дедуплицирует и сохраняет загрузку в
|
||||
// состоянии `catched`, сразу возвращая ответ. Вывод отображаемого имени
|
||||
// (медленный LLM) и добавление в qBittorrent — отдельный асинхронный шаг
|
||||
// worker'а (см. download-tracking).
|
||||
package ingest
|
||||
|
||||
import (
|
||||
@@ -11,21 +14,16 @@ import (
|
||||
|
||||
"git.vakhrushev.me/av/jellybit/internal/logctx"
|
||||
"git.vakhrushev.me/av/jellybit/internal/magnet"
|
||||
"git.vakhrushev.me/av/jellybit/internal/qbt"
|
||||
"git.vakhrushev.me/av/jellybit/internal/store"
|
||||
)
|
||||
|
||||
// capIngest — стадия приёма для поля capability в логах.
|
||||
const capIngest = "ingest"
|
||||
|
||||
// errCodeQbitAdd — error_code задачи, упавшей на добавлении источника в
|
||||
// qBittorrent (раздачи в qBittorrent нет, восстановлению не подлежит).
|
||||
const errCodeQbitAdd = "qbit_add"
|
||||
|
||||
// Store — нужная ingest часть хранилища.
|
||||
type Store interface {
|
||||
// FindActiveByInfohash — быстрый читающий дедуп-чек (до вызова LLM-namer);
|
||||
// авторитетная проверка — внутри CreateDownloadIfNoActive.
|
||||
// FindActiveByInfohash — быстрый читающий дедуп-чек; авторитетная проверка —
|
||||
// внутри CreateDownloadIfNoActive.
|
||||
FindActiveByInfohash(ctx context.Context, hashes ...string) (*store.Download, error)
|
||||
// CreateDownloadIfNoActive атомарно проверяет инвариант «одна активная
|
||||
// загрузка на infohash» и заводит задачу; вернувшаяся existing ≠ nil —
|
||||
@@ -34,50 +32,19 @@ type Store interface {
|
||||
// AddInfohashes доносит задаче недостающие хеши (guarded). Нужен на
|
||||
// быстром дедуп-пути, который не доходит до CreateDownloadIfNoActive.
|
||||
AddInfohashes(ctx context.Context, downloadID string, hashes []string) error
|
||||
SetDownloadState(ctx context.Context, id string, state store.State, errCode, errMsg string) error
|
||||
}
|
||||
|
||||
// QBittorrent — нужная ingest часть клиента qBittorrent.
|
||||
type QBittorrent interface {
|
||||
Add(ctx context.Context, ar qbt.AddRequest) error
|
||||
}
|
||||
|
||||
// Namer выводит человекочитаемое отображаемое имя торрента из контекста.
|
||||
// Пустой результат → имя в qBittorrent не задаём. nil → шаг пропускается.
|
||||
type Namer interface {
|
||||
DeriveName(ctx context.Context, contextText, hint string) string
|
||||
}
|
||||
|
||||
// Config — параметры добавления в qBittorrent.
|
||||
type Config struct {
|
||||
Category string
|
||||
SavePath string
|
||||
}
|
||||
|
||||
// Service — реализация приёма.
|
||||
// Service — реализация быстрого приёма.
|
||||
type Service struct {
|
||||
store Store
|
||||
qbt QBittorrent
|
||||
namer Namer
|
||||
cfg Config
|
||||
log *slog.Logger
|
||||
|
||||
// notifyFailed — опц. пинг автору о падении приёма (добавление в qBittorrent
|
||||
// не удалось). Closure, а не worker.Notifier: приёмное падение в qBit не
|
||||
// попадает в поллинг-цикл worker (раздачи нет), поэтому уведомляет ingest
|
||||
// сам; closure избавляет ядро приёма от зависимости на пакет worker.
|
||||
notifyFailed func(downloadID string)
|
||||
}
|
||||
|
||||
// New собирает сервис приёма. namer опционален (nil → отображаемое имя не
|
||||
// выводится; qBittorrent оставит своё).
|
||||
func New(st Store, qb QBittorrent, namer Namer, cfg Config, log *slog.Logger) *Service {
|
||||
return &Service{store: st, qbt: qb, namer: namer, cfg: cfg, log: log}
|
||||
// New собирает сервис приёма.
|
||||
func New(st Store, log *slog.Logger) *Service {
|
||||
return &Service{store: st, log: log}
|
||||
}
|
||||
|
||||
// SetFailureNotifier подключает пинг о падении приёма (до начала работы).
|
||||
func (s *Service) SetFailureNotifier(fn func(downloadID string)) { s.notifyFailed = fn }
|
||||
|
||||
// Request — входной запрос приёма.
|
||||
type Request struct {
|
||||
Source string // пока — magnet-ссылка
|
||||
@@ -92,8 +59,10 @@ type Result struct {
|
||||
Deduplicated bool // присоединились к уже активной задаче, нового добавления не было
|
||||
}
|
||||
|
||||
// Ingest принимает источник: извлекает infohash, дедуплицирует по активной
|
||||
// задаче, иначе заводит задачу и отдаёт источник в qBittorrent.
|
||||
// Ingest быстро принимает источник: извлекает infohash, синтезирует контекст из
|
||||
// полей ссылки, дедуплицирует по активной задаче, иначе сохраняет загрузку в
|
||||
// `catched` и сразу возвращает результат. Добавление в qBittorrent и вывод
|
||||
// имени выполняет worker (см. download-tracking).
|
||||
func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
source := strings.TrimSpace(req.Source)
|
||||
info, err := magnet.Parse(source)
|
||||
@@ -103,14 +72,12 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
}
|
||||
|
||||
// Scoped-логгер стадии приёма: download_id допишется после CreateDownload.
|
||||
// Кладём в ctx, чтобы внешние клиенты (qBittorrent, LLM-namer) дописывали
|
||||
// ключи корреляции к своим ext.*-записям сами.
|
||||
log := s.log.With("capability", capIngest, "infohash", info.Infohash)
|
||||
ctx = logctx.With(ctx, log)
|
||||
|
||||
// Быстрый дедуп-чек до дорогого LLM-namer; авторитетная (атомарная)
|
||||
// проверка — внутри CreateDownloadIfNoActive ниже. Дедуп — по ЛЮБОМУ из
|
||||
// хешей источника: гибридный magnet несёт и v1, и v2.
|
||||
// Быстрый дедуп-чек; авторитетная (атомарная) проверка — внутри
|
||||
// CreateDownloadIfNoActive ниже. Дедуп — по ЛЮБОМУ из хешей источника:
|
||||
// гибридный magnet несёт и v1, и v2.
|
||||
if existing, err := s.store.FindActiveByInfohash(ctx, info.Infohashes...); err != nil {
|
||||
return Result{}, fmt.Errorf("ingest: lookup active: %w", err)
|
||||
} else if existing != nil {
|
||||
@@ -118,26 +85,15 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
return s.attached(ctx, info, existing), nil
|
||||
}
|
||||
|
||||
// Отображаемое имя для списка qBit — best-effort: не валит приём.
|
||||
// Выводится синхронно (param rename действует только при добавлении) и
|
||||
// ДО CreateDownload, чтобы возможный медленный вызов LLM не расширял окно
|
||||
// «строка в БД есть, в qBittorrent ещё нет». Имя от строки БД не зависит.
|
||||
// Namer получает СЫРОЙ req.Context (+ dn-hint), не обогащённый: строки-факты
|
||||
// синтеза (Размер:/Трекер:) не должны становиться отображаемым именем.
|
||||
var rename string
|
||||
if s.namer != nil {
|
||||
rename = s.namer.DeriveName(ctx, req.Context, info.DisplayName)
|
||||
}
|
||||
|
||||
// Контекст распознавания дополняем фактами из полей самой magnet-ссылки
|
||||
// (dn/xl/tr/xs/kt) — без сети. Пользовательский текст идёт первым. Результат
|
||||
// уходит только в download.Context (его читают recognition и веб-UI).
|
||||
// (dn/xl/tr/xs/kt) — без сети. Пользовательский текст идёт первым.
|
||||
// DisplayName пуст: имя выведет worker на шаге добавления (rename действует
|
||||
// только при добавлении, а тут медленный LLM в пути ответа недопустим).
|
||||
d := &store.Download{
|
||||
SourceType: store.SourceMagnet,
|
||||
SourceRef: source,
|
||||
DisplayName: rename, // то же имя, что уходит в qBittorrent (rename); заголовок в веб-UI
|
||||
Context: mergeContext(req.Context, info.Context()),
|
||||
State: store.StateDownloading,
|
||||
SourceType: store.SourceMagnet,
|
||||
SourceRef: source,
|
||||
Context: mergeContext(req.Context, info.Context()),
|
||||
State: store.StateCatched,
|
||||
}
|
||||
// Все хеши из magnet (гибридный несёт v1 и v2); kind store выведет по длине.
|
||||
existing, err := s.store.CreateDownloadIfNoActive(ctx, d, info.Infohashes)
|
||||
@@ -156,38 +112,12 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
Deduplicated: true,
|
||||
}, nil
|
||||
}
|
||||
id := d.ID
|
||||
log = log.With("download_id", id)
|
||||
ctx = logctx.With(ctx, log)
|
||||
|
||||
addErr := s.qbt.Add(ctx, qbt.AddRequest{
|
||||
URLs: []string{source},
|
||||
Category: s.cfg.Category,
|
||||
SavePath: s.cfg.SavePath,
|
||||
Rename: rename,
|
||||
})
|
||||
if addErr != nil {
|
||||
// Граница доменной операции приёма: логируем исход один раз (ERROR).
|
||||
// Поведение самого вызова qBittorrent уже залогировал клиент (ext.*) —
|
||||
// это разные факты, не дубль.
|
||||
log.Error("download accept failed", "error", addErr)
|
||||
// Задача уже в БД — помечаем failed, чтобы worker её не подхватил.
|
||||
if setErr := s.store.SetDownloadState(ctx, id, store.StateFailed, errCodeQbitAdd, addErr.Error()); setErr != nil {
|
||||
log.Error("mark download failed after qbit error failed", "error", setErr)
|
||||
} else if s.notifyFailed != nil {
|
||||
// Это падение минует worker.transition (раздачи в qBit нет) — уведомляем
|
||||
// сами, чтобы приёмные провалы тоже доходили до автора.
|
||||
go s.notifyFailed(id)
|
||||
}
|
||||
return Result{DownloadID: id, Infohashes: info.Infohashes, State: store.StateFailed},
|
||||
fmt.Errorf("ingest: add to qbittorrent: %w", addErr)
|
||||
}
|
||||
|
||||
log.Info("download accepted", "category", s.cfg.Category)
|
||||
log.Info("download catched", "download_id", d.ID)
|
||||
return Result{
|
||||
DownloadID: id,
|
||||
DownloadID: d.ID,
|
||||
Infohashes: info.Infohashes,
|
||||
State: store.StateDownloading,
|
||||
State: store.StateCatched,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user