Приём/UI: пачка фиксов границ и парсинга (F7–F10, N2)
Пять независимых bugfix'ов из ревью приёма (docs/backlog/review-f7-f10-ingest-ui-fixes.md): - F7: oversized .torrent через веб отдавал 500. Введён sentinel ingest.ErrTorrentTooLarge, classifyErr транслирует его в 400. - F8: гонка fast-path attach с cancel. Пред-рид FindReingestBlockingByInfohash больше не короткозамыкает активную запись — авторитетное дедуп-решение принимает CreateDownloadIfNoActive под BEGIN IMMEDIATE; короткозамыкание оставлено только для терминальных desync-записей (target_missing/orphaned). F6-апгрейд сохранён. - F9: magnet — регистронезависимый URN-префикс xt (RFC 2141); tgbot.ParseMessage срезает хвостовую пунктуацию, приклеенную жадным matchем. - F10: cap контекста до 16 KiB в ingest.Ingest (единственное место слияния — покрывает все транспорты), рунобезопасная обрезка + маркер. - N2: httpapi.shorten режет по рунам, не байтам — кириллица не рвётся в U+FFFD. Добавлены юнит-тесты на каждое исправленное поведение. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -8,9 +8,11 @@ package ingest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.vakhrushev.me/av/jellybit/internal/logctx"
|
||||
"git.vakhrushev.me/av/jellybit/internal/magnet"
|
||||
@@ -89,17 +91,21 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
log := s.log.With("capability", capIngest, "infohash", src.infohashes[0])
|
||||
ctx = logctx.With(ctx, log)
|
||||
|
||||
// Быстрый дедуп-чек по ЛЮБОМУ из хешей источника (гибридный несёт и v1, и v2):
|
||||
// Быстрый пред-рид по ЛЮБОМУ из хешей источника (гибридный несёт и v1, и v2):
|
||||
// активная задача ЛИБО удерживающая источник desync-запись
|
||||
// (target_missing/orphaned) блокируют повторный приём. Для активной
|
||||
// авторитетная (атомарная) проверка — внутри CreateDownloadIfNoActive ниже;
|
||||
// desync-ветка сюда и завершается (в active-гард desync не заводим, чтобы не
|
||||
// размыть инвариант «≤1 активной»).
|
||||
// (target_missing/orphaned) блокируют повторный приём. Здесь короткозамыкаем
|
||||
// ТОЛЬКО desync-запись (терминальную): её CreateDownloadIfNoActive не увидит
|
||||
// (тот проверяет лишь активных), а состояния она не меняет. Активную же НЕ
|
||||
// короткозамыкаем — этот чек без транзакции, и в гонке с параллельным cancel
|
||||
// вернул бы stale «уже в работе» при пустом активном множестве. Авторитетное
|
||||
// дедуп-решение по активной примет CreateDownloadIfNoActive под BEGIN IMMEDIATE.
|
||||
if existing, err := s.store.FindReingestBlockingByInfohash(ctx, src.infohashes...); err != nil {
|
||||
// Инфраструктурный сбой (БД) — операция приёма не выполнена: ERROR.
|
||||
log.Error("ingest failed", "stage", "lookup-blocking", "error", err)
|
||||
return Result{}, fmt.Errorf("ingest: lookup blocking: %w", err)
|
||||
} else if existing != nil {
|
||||
} else if existing != nil && existing.State.IsTerminal() {
|
||||
// FindReingestBlockingByInfohash отдаёт терминальную запись только из
|
||||
// удерживающих desync-состояний (target_missing/orphaned) — присоединяемся.
|
||||
log.Info("download attached", "download_id", existing.ID, "state", existing.State)
|
||||
return s.attached(ctx, src, existing), nil
|
||||
}
|
||||
@@ -111,7 +117,7 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
d := &store.Download{
|
||||
SourceType: src.sourceType,
|
||||
SourceRef: src.sourceRef,
|
||||
Context: mergeContext(req.Context, src.synthContext),
|
||||
Context: capContext(mergeContext(req.Context, src.synthContext)),
|
||||
State: store.StateCatched,
|
||||
}
|
||||
// Все хеши источника (гибрид несёт v1 и v2); kind store выведет по длине.
|
||||
@@ -144,6 +150,12 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
|
||||
// файлов → много piece-хешей) отсекаются здесь.
|
||||
const MaxTorrentSize = 8 << 20 // 8 MiB
|
||||
|
||||
// ErrTorrentTooLarge — принятый `.torrent` превышает MaxTorrentSize. Это промах
|
||||
// ввода пользователя (норма, не сбой сервера), поэтому транспорт транслирует его
|
||||
// в 400, а не 500 (веб MaxBytesReader пропускает файлы чуть больше лимита —
|
||||
// отсекает уже приём). Проверяется через errors.Is.
|
||||
var ErrTorrentTooLarge = errors.New("torrent too large")
|
||||
|
||||
// parsedSource — нормализованный источник приёма (magnet или .torrent).
|
||||
type parsedSource struct {
|
||||
sourceType store.SourceType
|
||||
@@ -159,7 +171,7 @@ type parsedSource struct {
|
||||
func (s *Service) parse(req Request) (parsedSource, error) {
|
||||
if len(req.TorrentData) > 0 {
|
||||
if len(req.TorrentData) > MaxTorrentSize {
|
||||
return parsedSource{}, fmt.Errorf("ingest: torrent too large: %d > %d bytes", len(req.TorrentData), MaxTorrentSize)
|
||||
return parsedSource{}, fmt.Errorf("ingest: torrent too large: %d > %d bytes: %w", len(req.TorrentData), MaxTorrentSize, ErrTorrentTooLarge)
|
||||
}
|
||||
info, err := torrent.Parse(req.TorrentData)
|
||||
if err != nil {
|
||||
@@ -195,6 +207,38 @@ func (s *Service) parse(req Request) (parsedSource, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
// MaxContextSize — предел размера контекста распознавания (пользовательский текст
|
||||
// + синтез из полей источника). Кап здесь, на единственном месте слияния,
|
||||
// покрывает все транспорты: REST ограничен телом (64 KiB), Telegram — лимитом
|
||||
// подписи, но веб-форма (multipart-бюджет на всё тело) иначе пропустила бы
|
||||
// мегабайты в поле context → в БД, рендер карточки и LLM-промпты.
|
||||
const MaxContextSize = 16 << 10 // 16 KiB
|
||||
|
||||
// contextTruncMarker дописывается к усечённому контексту как явный маркер.
|
||||
const contextTruncMarker = "\n…[контекст усечён]"
|
||||
|
||||
// capContext ограничивает контекст MaxContextSize байтами, обрезая по границе
|
||||
// руны (кириллица — 2 байта/руна; обрезка посреди руны дала бы U+FFFD) и добавляя
|
||||
// маркер усечения. Пустой/короткий контекст возвращается как есть.
|
||||
func capContext(s string) string {
|
||||
if len(s) <= MaxContextSize {
|
||||
return s
|
||||
}
|
||||
return trimToRune(s[:MaxContextSize]) + contextTruncMarker
|
||||
}
|
||||
|
||||
// trimToRune отбрасывает незавершённую многобайтовую руну на конце строки
|
||||
// (результат обрезки по фиксированному числу байт), не трогая корректный хвост.
|
||||
func trimToRune(s string) string {
|
||||
for len(s) > 0 {
|
||||
if r, size := utf8.DecodeLastRuneInString(s); r != utf8.RuneError || size > 1 {
|
||||
break
|
||||
}
|
||||
s = s[:len(s)-1]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// mergeContext склеивает контекст от транспорта с синтезом из полей magnet:
|
||||
// пользовательский текст идёт первым, затем факты из ссылки. Пустые части
|
||||
// опускаются; при пустых обеих — пустая строка (пустой контекст допустим).
|
||||
|
||||
@@ -2,6 +2,7 @@ package ingest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"strings"
|
||||
@@ -53,6 +54,30 @@ func (f *fakeStore) UpgradeCatchedMagnetToTorrent(_ context.Context, id string,
|
||||
return f.upgradeUp, nil
|
||||
}
|
||||
|
||||
// raceStore моделирует гонку F8: пред-рид FindReingestBlockingByInfohash видит
|
||||
// активную запись (blocking), но create-гард CreateDownloadIfNoActive её уже не
|
||||
// находит (в параллели отменена) и заводит свежую задачу.
|
||||
type raceStore struct {
|
||||
blocking *store.Download
|
||||
created []store.Download
|
||||
}
|
||||
|
||||
func (r *raceStore) FindReingestBlockingByInfohash(_ context.Context, _ ...string) (*store.Download, error) {
|
||||
return r.blocking, nil
|
||||
}
|
||||
|
||||
func (r *raceStore) CreateDownloadIfNoActive(_ context.Context, d *store.Download, _ []string, _ []byte) (*store.Download, error) {
|
||||
d.ID = ident.NewID()
|
||||
r.created = append(r.created, *d)
|
||||
return nil, nil // активной уже нет — создаём новую
|
||||
}
|
||||
|
||||
func (r *raceStore) AddInfohashes(_ context.Context, _ string, _ []string) error { return nil }
|
||||
|
||||
func (r *raceStore) UpgradeCatchedMagnetToTorrent(_ context.Context, _ string, _ []byte) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
func newService(st Store) *Service {
|
||||
return New(st, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
}
|
||||
@@ -207,6 +232,66 @@ func TestIngestDedupTopsUpHashes(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// F8: пред-рид FindReingestBlockingByInfohash увидел активную задачу, но к моменту
|
||||
// создания она отменена (гонка с cancel). Активную запись пред-рид НЕ
|
||||
// короткозамыкает — авторитетное дедуп-решение принимает CreateDownloadIfNoActive
|
||||
// под BEGIN IMMEDIATE: активной больше нет → заводим свежую задачу, а не
|
||||
// возвращаем stale Deduplicated «уже в работе».
|
||||
func TestIngestActivePreReadNotShortCircuited(t *testing.T) {
|
||||
stale := &store.Download{ID: "01hzzzstale00000000000000000", State: store.StateCatched}
|
||||
fs := &raceStore{blocking: stale} // пред-рид видит активную; create-гард — уже нет
|
||||
res, err := newService(fs).Ingest(context.Background(), Request{Source: sampleMagnet})
|
||||
if err != nil {
|
||||
t.Fatalf("Ingest: %v", err)
|
||||
}
|
||||
if res.Deduplicated {
|
||||
t.Errorf("активный пред-рид не должен коротко замыкать дедуп: %+v", res)
|
||||
}
|
||||
if res.DownloadID == stale.ID || res.State != store.StateCatched {
|
||||
t.Errorf("ожидалась свежая задача, а не stale: %+v", res)
|
||||
}
|
||||
if len(fs.created) != 1 {
|
||||
t.Errorf("должна быть создана новая задача, created=%d", len(fs.created))
|
||||
}
|
||||
}
|
||||
|
||||
// F7: oversized `.torrent` — доменная ошибка размера класса ErrTorrentTooLarge
|
||||
// (транспорт транслирует в 400, а не 500). Задача не заводится.
|
||||
func TestIngestRejectsOversizedTorrent(t *testing.T) {
|
||||
fs := &fakeStore{}
|
||||
big := make([]byte, MaxTorrentSize+1)
|
||||
_, err := newService(fs).Ingest(context.Background(), Request{TorrentData: big})
|
||||
if !errors.Is(err, ErrTorrentTooLarge) {
|
||||
t.Fatalf("err = %v, want ErrTorrentTooLarge", err)
|
||||
}
|
||||
if len(fs.created) != 0 {
|
||||
t.Error("не должно быть записи задачи")
|
||||
}
|
||||
}
|
||||
|
||||
// F10: контекст из веб-формы может быть огромным (multipart-бюджет на всё тело) —
|
||||
// Ingest режет его до MaxContextSize по границе руны (без U+FFFD) и метит маркером.
|
||||
func TestIngestCapsContext(t *testing.T) {
|
||||
// «Ё» — 2 байта; ASCII-префикс сдвигает границу MaxContextSize на нечётный
|
||||
// байт, чтобы обрезка s[:MaxContextSize] пришлась ВНУТРЬ двухбайтовой руны —
|
||||
// тогда trimToRune реально срабатывает (иначе граница попадёт между рунами).
|
||||
huge := "x" + strings.Repeat("Ё", MaxContextSize)
|
||||
fs := &fakeStore{}
|
||||
if _, err := newService(fs).Ingest(context.Background(), Request{Source: sampleMagnet, Context: huge}); err != nil {
|
||||
t.Fatalf("Ingest: %v", err)
|
||||
}
|
||||
got := fs.created[0].Context
|
||||
if len(got) > MaxContextSize+len(contextTruncMarker) {
|
||||
t.Errorf("контекст не ограничен: %d байт", len(got))
|
||||
}
|
||||
if !strings.HasSuffix(got, contextTruncMarker) {
|
||||
t.Errorf("нет маркера усечения: …%q", got[max(0, len(got)-40):])
|
||||
}
|
||||
if strings.ContainsRune(got, '�') {
|
||||
t.Error("обрезка порвала руну (U+FFFD)")
|
||||
}
|
||||
}
|
||||
|
||||
func TestIngestRejectsNonMagnet(t *testing.T) {
|
||||
fs := &fakeStore{}
|
||||
if _, err := newService(fs).Ingest(context.Background(), Request{Source: "https://example.com/x.torrent"}); err == nil {
|
||||
|
||||
Reference in New Issue
Block a user