Приём: дедуп по target_missing/orphaned + стоп-кран «Закрыть»
Два дубля-близнеца на один инфохэш рождались, когда повторный приём
попадал на запись в target_missing: дедуп искал только активную задачу,
а target_missing терминален → заводилась новая загрузка, воркер усыновлял
уже присутствующий торрент и раскладывал его.
- Приём: критерий дедупа расширен до «блокирующей повторный приём» =
активные ∪ {target_missing, orphaned}. Повторный приём такого инфохэша
привязывается к существующей записи (спящей, без обращения к qBittorrent),
а не плодит близнеца. Прочие терминальные (done/cancelled/failed/reverted/
deleted) повторный приём не блокируют — осознанная свежая попытка. Новый
read-метод FindReingestBlockingByInfohash (приоритет активной над desync);
общий active-гард не тронут.
- Команда «Закрыть» (Dismiss) — универсальный стоп-кран из любого состояния,
кроме deleted → cancelled (error_code=user_dismiss). Только меняет статус:
файлы (в т.ч. хардлинки done/orphaned) и раздачу qBittorrent не трогает,
в отличие от «Удалить». Веб — danger-зона внизу страницы; Telegram —
кнопка с подтверждением; из cancelled — идемпотентный no-op.
- Транспорты при дедупе на desync-запись сообщают адресно (target_missing —
привязать заново/закрыть; orphaned — закрыть и добавить заново); веб при
дедупе ведёт на страницу существующей записи.
Спеки: ingest (дедуп), state-reconciliation (стоп-кран); граф переходов
допополнен рёбрами <терминал>→cancelled. OpenSpec change
dedup-target-missing-and-dismiss заархивирован.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -78,9 +78,14 @@ func (s State) IsTerminal() bool {
|
||||
// (ActivateIfNoOtherActive): гейт графа ортогонален гарду терминальности в
|
||||
// setState — граф говорит «ребро есть», гард «но не мимо ActivateIfNoOtherActive».
|
||||
// Так, failed → downloading объявлено, но обычным SetDownloadState отклоняется.
|
||||
// - cancelled/deferred — легальная цель из КАЖДОГО не-терминального состояния
|
||||
// (Cancel/Defer проверяют лишь IsTerminal); инвариант закреплён тестом, а не
|
||||
// ручной аккуратностью.
|
||||
// - deferred — легальная цель из КАЖДОГО не-терминального состояния (Defer
|
||||
// проверяет лишь IsTerminal); инвариант закреплён тестом, а не ручной
|
||||
// аккуратностью.
|
||||
// - cancelled — легальная цель из ЛЮБОГО состояния, кроме deleted: помимо
|
||||
// Cancel из не-терминальных её даёт универсальный стоп-кран Dismiss, доступный
|
||||
// и из терминальных (done/failed/reverted/target_missing/orphaned) — только
|
||||
// смена статуса, файлы/раздачу не трогает (см. state-reconciliation «Ручное
|
||||
// закрытие загрузки»). deleted строго терминален и цель cancelled не получает.
|
||||
//
|
||||
// Правка воркера, вводящая новое ребро, ОБЯЗАНА отразить его здесь — иначе
|
||||
// setState отклонит переход (0 строк UPDATE → ошибка).
|
||||
@@ -91,14 +96,14 @@ var allowedTransitions = map[State][]State{
|
||||
StateRecognizing: {StateLinking, StateReview, StateCancelled, StateDeferred},
|
||||
StateReview: {StateLinking, StateRecognizing, StateCancelled, StateDeferred, StateOrphaned, StateDeleted},
|
||||
StateLinking: {StateDone, StateReview, StateFailed, StateCancelled, StateDeferred},
|
||||
StateDone: {StateReverted, StateTargetMissing, StateOrphaned, StateDeleted},
|
||||
StateDone: {StateReverted, StateTargetMissing, StateOrphaned, StateDeleted, StateCancelled},
|
||||
StateDeferred: {StateLinking, StateRecognizing, StateCancelled, StateOrphaned, StateDeleted},
|
||||
StateStuck: {StateDownloading, StateCompleted, StateCancelled, StateDeferred},
|
||||
StateFailed: {StateDownloading, StateCompleted},
|
||||
StateReverted: {StateRecognizing, StateOrphaned, StateDeleted},
|
||||
StateFailed: {StateDownloading, StateCompleted, StateCancelled},
|
||||
StateReverted: {StateRecognizing, StateOrphaned, StateDeleted, StateCancelled},
|
||||
StateCancelled: {StateRecognizing, StateOrphaned, StateDeleted},
|
||||
StateTargetMissing: {StateRecognizing, StateDone, StateOrphaned, StateDeleted},
|
||||
StateOrphaned: {StateDone, StateTargetMissing, StateDeleted},
|
||||
StateTargetMissing: {StateRecognizing, StateDone, StateOrphaned, StateDeleted, StateCancelled},
|
||||
StateOrphaned: {StateDone, StateTargetMissing, StateDeleted, StateCancelled},
|
||||
StateDeleted: nil, // окончательно терминально: сверка его не переоценивает
|
||||
}
|
||||
|
||||
@@ -616,6 +621,68 @@ func (s *Store) FindActiveByInfohash(ctx context.Context, hashes ...string) (*Do
|
||||
return d, nil
|
||||
}
|
||||
|
||||
// reingestHoldingStates — desync-состояния, которые удерживают источник ради
|
||||
// незакрытого намерения и потому БЛОКИРУЮТ повторный приём наравне с активными:
|
||||
// target_missing (источник жив, ждёт relink) и orphaned (источник пропал, запись
|
||||
// держит претензию на последнюю копию). Прочие терминальные (done/cancelled/
|
||||
// failed/reverted/deleted) повторный приём НЕ блокируют — это осознанная свежая
|
||||
// попытка. «Блокирующие» = активные (не-терминальные) ∪ reingestHoldingStates.
|
||||
var reingestHoldingStates = []State{StateTargetMissing, StateOrphaned}
|
||||
|
||||
// FindReingestBlockingByInfohash возвращает задачу, блокирующую повторный приём
|
||||
// любого из hashes: активную (строго не-терминальную) ЛИБО удерживающую источник
|
||||
// desync-запись (target_missing/orphaned), приоритет — активной. Либо (nil, nil).
|
||||
// Читающая основа расширенного дедупа приёма (пред-рид ДО создания); инвариант
|
||||
// «≤1 активной на infohash» держит отдельный active-гард CreateDownloadIfNoActive,
|
||||
// в который desync-состояния НЕ заводятся.
|
||||
func (s *Store) FindReingestBlockingByInfohash(ctx context.Context, hashes ...string) (*Download, error) {
|
||||
norm := normalizeHashes(hashes)
|
||||
// Активная имеет приоритет: если по хешу есть и активная, и desync-запись
|
||||
// (инвариант это допускает), присоединяемся к активной.
|
||||
d, err := findActiveByInfohash(ctx, s.DB, norm, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if d == nil {
|
||||
d, err = findByInfohashInStates(ctx, s.DB, norm, reingestHoldingStates)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if d != nil {
|
||||
if err := attachInfohashesOne(ctx, s.DB, d); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return d, nil
|
||||
}
|
||||
|
||||
// findByInfohashInStates — выборка «задача по любому из хешей в одном из states»
|
||||
// (позитивный фильтр `state IN (...)`, в отличие от findActiveByInfohash с
|
||||
// `NOT IN terminalStates`). hashes уже нормализованы; хеши найденной загрузки НЕ
|
||||
// подгружаются. Пустые hashes/states → (nil, nil).
|
||||
func findByInfohashInStates(ctx context.Context, q sqlx.QueryerContext, hashes []string, states []State) (*Download, error) {
|
||||
if len(hashes) == 0 || len(states) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
var args []any
|
||||
hashPh := placeholders(&args, hashes)
|
||||
statePh := placeholders(&args, states)
|
||||
query := `SELECT download.* FROM download
|
||||
JOIN download_infohash dh ON dh.download_id = download.id
|
||||
WHERE dh.infohash IN (` + hashPh + `) AND download.state IN (` + statePh + `)
|
||||
ORDER BY download.id DESC LIMIT 1`
|
||||
var d Download
|
||||
err := sqlx.GetContext(ctx, q, &d, query, args...)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find by infohash in states: %w", err)
|
||||
}
|
||||
return &d, nil
|
||||
}
|
||||
|
||||
// findActiveByInfohash — общая выборка «активная задача по любому из хешей»
|
||||
// (для guarded-методов — внутри их транзакции). hashes уже нормализованы;
|
||||
// excludeID исключает саму проверяемую задачу (она может быть активной,
|
||||
|
||||
@@ -260,6 +260,71 @@ func TestFindActiveByInfohash_DesyncStatesNotActive(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// Дедуп повторного приёма блокируют не только активные, но и удерживающие
|
||||
// источник desync-записи (target_missing/orphaned): по ним приём привязывается к
|
||||
// существующей, а не плодит близнеца. Прочие терминальные (done/cancelled/failed/
|
||||
// reverted/deleted) повторный приём НЕ блокируют — осознанная свежая попытка.
|
||||
func TestFindReingestBlockingByInfohash(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
blocking := []State{
|
||||
StateCatched, StateDownloading, StateReview, // активные (примеры)
|
||||
StateTargetMissing, StateOrphaned, // desync, удерживающие источник
|
||||
}
|
||||
for _, s := range blocking {
|
||||
t.Run("blocking/"+string(s), func(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
ih := hashN(1)
|
||||
id := mustCreate(t, st, ih)
|
||||
forceState(t, st, id, s)
|
||||
d, err := st.FindReingestBlockingByInfohash(ctx, ih)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if d == nil || d.ID != id {
|
||||
t.Fatalf("%s должна блокировать приём, получили %v", s, d)
|
||||
}
|
||||
if len(d.Infohashes) != 1 {
|
||||
t.Fatalf("хеши не подгружены: %v", d.Infohashes)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
nonBlocking := []State{
|
||||
StateDone, StateCancelled, StateFailed, StateReverted, StateDeleted,
|
||||
}
|
||||
for _, s := range nonBlocking {
|
||||
t.Run("non-blocking/"+string(s), func(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
ih := hashN(2)
|
||||
id := mustCreate(t, st, ih)
|
||||
forceState(t, st, id, s)
|
||||
d, err := st.FindReingestBlockingByInfohash(ctx, ih)
|
||||
if err != nil || d != nil {
|
||||
t.Fatalf("%s блокировать приём не должна, получили (%v,%v)", s, d, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("priority-active-over-desync", func(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
ih := hashN(3)
|
||||
// Старая запись ушла в target_missing (terminal освобождает хеш), затем по
|
||||
// тому же хешу завелась новая активная — инвариант «≤1 активной» это
|
||||
// допускает. Дедуп обязан присоединиться к активной, а не к desync.
|
||||
oldID := mustCreate(t, st, ih)
|
||||
forceState(t, st, oldID, StateTargetMissing)
|
||||
newID := mustCreate(t, st, ih)
|
||||
d, err := st.FindReingestBlockingByInfohash(ctx, ih)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if d == nil || d.ID != newID {
|
||||
t.Fatalf("ожидалась активная %s (приоритет над desync %s), получили %v", newID, oldID, d)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Терминальное состояние освобождает infohash: тот же хеш заводится заново
|
||||
// новой задачей (повторная закачка спустя время) — активность выводится
|
||||
// только из state.
|
||||
|
||||
@@ -74,19 +74,12 @@ func TestTransitionGraphWellFormed(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// Инвариант generic-команд Cancel/Defer: cancelled и deferred — легальная цель
|
||||
// из КАЖДОГО не-терминального состояния (кроме самого deferred для deferred —
|
||||
// это самопереход). Ловит класс дыры «забыли состояние» (напр. linking после
|
||||
// краха процесса).
|
||||
func TestCancelDeferReachableFromEveryNonTerminal(t *testing.T) {
|
||||
// Инвариант Defer: deferred — легальная цель из КАЖДОГО не-терминального
|
||||
// состояния (кроме самого deferred — это самопереход). Ловит класс дыры «забыли
|
||||
// состояние» (напр. linking после краха процесса).
|
||||
func TestDeferReachableFromEveryNonTerminal(t *testing.T) {
|
||||
for _, s := range allStates {
|
||||
if s.IsTerminal() {
|
||||
continue
|
||||
}
|
||||
if !slices.Contains(transitionSources[StateCancelled], s) {
|
||||
t.Errorf("%s → cancelled не легально (Cancel допускает любое не-терминальное)", s)
|
||||
}
|
||||
if s == StateDeferred {
|
||||
if s.IsTerminal() || s == StateDeferred {
|
||||
continue // deferred → deferred покрыт самопереходом
|
||||
}
|
||||
if !slices.Contains(transitionSources[StateDeferred], s) {
|
||||
@@ -95,6 +88,22 @@ func TestCancelDeferReachableFromEveryNonTerminal(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// Инвариант универсального стоп-крана Dismiss: cancelled — легальная цель из
|
||||
// ЛЮБОГО состояния, кроме deleted (строго терминален) и самого cancelled
|
||||
// (самопереход). Шире инварианта Defer: покрывает и терминальные
|
||||
// done/failed/reverted/target_missing/orphaned. НЕ объединять с проверкой
|
||||
// deferred — у них разные множества источников.
|
||||
func TestCancelledReachableFromEveryStateButDeleted(t *testing.T) {
|
||||
for _, s := range allStates {
|
||||
if s == StateDeleted || s == StateCancelled {
|
||||
continue // deleted строго терминален; cancelled → cancelled — самопереход
|
||||
}
|
||||
if !slices.Contains(transitionSources[StateCancelled], s) {
|
||||
t.Errorf("%s → cancelled не легально (Dismiss/Cancel допускают любое состояние, кроме deleted)", s)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Объявленные не-revive рёбра проходят через SetDownloadState.
|
||||
func TestSetStateAllowsDeclaredEdges(t *testing.T) {
|
||||
edges := []struct{ from, to State }{
|
||||
@@ -110,6 +119,14 @@ func TestSetStateAllowsDeclaredEdges(t *testing.T) {
|
||||
{StateDone, StateReverted},
|
||||
{StateStuck, StateCancelled},
|
||||
{StateReview, StateDeferred},
|
||||
// Стоп-кран Dismiss: терминал → cancelled идёт обычным SetDownloadState
|
||||
// (цель cancelled терминальна → гард терминальности не мешает, revive не
|
||||
// нужен).
|
||||
{StateDone, StateCancelled},
|
||||
{StateFailed, StateCancelled},
|
||||
{StateReverted, StateCancelled},
|
||||
{StateTargetMissing, StateCancelled},
|
||||
{StateOrphaned, StateCancelled},
|
||||
}
|
||||
for i, e := range edges {
|
||||
st := newTestStore(t)
|
||||
|
||||
Reference in New Issue
Block a user