av/jellybit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Приём: пропажа источника у активной загрузки → failed(source_gone) (MAJOR-3)

Browse Source 
Раздача активной (downloading) загрузки, исчезнувшая из qBittorrent (удалил
пользователь/другой клиент), делала задачу вечным зомби: поллинг промахивался
по torrentFor, писал Warn и continue каждый тик — состояние не менялось,
уведомления и телеметрии не было, checkTimeouts без торрента не срабатывал.
Пропажей источника у downloading не владел никто (сверка рассинхрона покрывает
только done/target_missing/orphaned, восстановление — failed/stuck).

Активный цикл Poll теперь применяет тот же дебаунс пропажи источника, что и
сверка рассинхрона (source_miss_count / source_missing_threshold): после порога
подряд идущих промахов задача уходит downloading → failed с distinct error_code
source_gone и уведомлением. До порога транзиентная недоступность qBit
(рестарт демона) задачу не роняет. source_gone восстановлению сверкой не
подлежит (удаление намеренно), но штатно retriable — Retry заново отдаёт
сохранённый источник; Retry сбрасывает source_miss_count, чтобы вернувшаяся
задача получила полное грейс-окно, а не упала снова на ближайшем тике.

Ребро downloading → failed уже было в графе, миграций/полей БД нет. Спека
download-tracking дополнена требованием, диаграмма workflow.md — ребром.
Change downloading-source-gone заархивирован.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
av
2026-07-08 18:05:04 +03:00
co-authored by Claude Opus 4.8
parent bfd469bd43
commit 2a5a65f2d5
13 changed files with 524 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/worker/reconcile_test.go
+8 -4
View File
@@ -151,13 +151,17 @@ func TestReconcileDebounce(t *testing.T) {
}
func TestReconcileSkipsActiveStates(t *testing.T) {
// downloading сверкой не трогаем, даже если раздачи нет в qBittorrent.
f := newReconcileFixture(t, store.StateDownloading, false, true)
// Матрица сверки (reconcileDesync) не трогает активные состояния, даже если
// раздачи нет в qBittorrent. Берём review (не downloading): активный цикл
// поллинга листает только downloading, поэтому review остаётся нетронутым
// — пропажу источника у downloading отдельно ведёт правило source_gone
// (см. recovery_test: TestSourceGone*).
f := newReconcileFixture(t, store.StateReview, false, true)
if err := f.w.Poll(context.Background()); err != nil {
t.Fatalf("Poll: %v", err)
}
if got := f.st.downloads["1"].State; got != store.StateDownloading {
t.Errorf("state = %q, want downloading (сверка не трогает активные)", got)
if got := f.st.downloads["1"].State; got != store.StateReview {
t.Errorf("state = %q, want review (сверка не трогает активные)", got)
}
}
internal/worker/recovery_test.go
+134
View File
@@ -260,3 +260,137 @@ func TestStallIgnoresFutureLastActivity(t *testing.T) {
t.Errorf("last_activity из будущего игнорируется (фолбэк на возраст 5ч > stuck_after 1ч → stuck), got %q", got)
}
}
// --- MAJOR-3: пропажа источника у активной (downloading) загрузки ---
// Источник активной загрузки устойчиво пропал из qBittorrent → задача уходит в
// failed(source_gone) с уведомлением, а не остаётся вечным зомби.
func TestSourceGoneMarksFailed(t *testing.T) {
const ih = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
st := oneDownloading(ih, timeRecent)
w := newTestWorker(st, &fakeQbt{torrents: nil}) // раздачи нет
w.cfg.SourceMissingThreshold = 1 // помечаем при первой же пропаже
n := &recordingNotifier{ch: make(chan notifyEvent, 4)}
w.SetNotifier(n)
if err := w.Poll(context.Background()); err != nil {
t.Fatalf("Poll: %v", err)
}
d := st.downloads["1"]
if d.State != store.StateFailed {
t.Fatalf("state = %q, want failed", d.State)
}
if got := d.ErrorCode.String; got != errCodeSourceGone {
t.Errorf("error_code = %q, want %q", got, errCodeSourceGone)
}
if e := waitNotify(t, n); e.ev != EventFailed {
t.Errorf("пинг: ev=%v, want failed", e.ev)
}
}
// Кратковременная пропажа (меньше порога) не роняет задачу — дебаунс терпит
// транзиентную недоступность qBittorrent (рестарт демона).
func TestSourceGoneDebounced(t *testing.T) {
const ih = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
st := oneDownloading(ih, timeRecent)
w := newTestWorker(st, &fakeQbt{torrents: nil})
w.cfg.SourceMissingThreshold = 3
for i := 1; i <= 2; i++ {
if err := w.Poll(context.Background()); err != nil {
t.Fatalf("Poll %d: %v", i, err)
}
if got := st.downloads["1"].State; got != store.StateDownloading {
t.Fatalf("tick %d: state = %q, want downloading (до порога)", i, got)
}
if got := st.downloads["1"].SourceMissCount; got != i {
t.Errorf("tick %d: miss = %d, want %d", i, got, i)
}
}
if err := w.Poll(context.Background()); err != nil { // третий промах
t.Fatalf("Poll 3: %v", err)
}
if got := st.downloads["1"].State; got != store.StateFailed {
t.Fatalf("tick 3: state = %q, want failed (порог достигнут)", got)
}
if got := st.downloads["1"].ErrorCode.String; got != errCodeSourceGone {
t.Errorf("error_code = %q, want %q", got, errCodeSourceGone)
}
}
// Возврат раздачи до порога сбрасывает счётчик промахов и задача ведётся
// обычной сверкой (не падает).
func TestSourceGoneResetOnReturn(t *testing.T) {
const ih = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
st := oneDownloading(ih, timeRecent)
qb := &fakeQbt{torrents: nil}
w := newTestWorker(st, qb)
w.cfg.SourceMissingThreshold = 3
if err := w.Poll(context.Background()); err != nil { // промах 1
t.Fatalf("Poll miss: %v", err)
}
if got := st.downloads["1"].SourceMissCount; got != 1 {
t.Fatalf("miss = %d, want 1", got)
}
// Раздача вернулась (свежий metaDL — не падает по таймауту).
qb.torrents = []qbt.Torrent{{Hash: ih, State: "metaDL", AddedOn: addedRecent}}
if err := w.Poll(context.Background()); err != nil {
t.Fatalf("Poll return: %v", err)
}
if got := st.downloads["1"].State; got != store.StateDownloading {
t.Errorf("state = %q, want downloading (источник вернулся)", got)
}
if got := st.downloads["1"].SourceMissCount; got != 0 {
t.Errorf("miss = %d, want 0 (сброс)", got)
}
}
// source_gone восстановлению сверкой не подлежит: даже если раздача снова
// появилась и продвинулась, reconcileRecovery её не воскрешает — нужен ручной
// Retry.
func TestSourceGoneNotAutoRecovered(t *testing.T) {
const ih = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
st := oneFailed(store.StateFailed, errCodeSourceGone, ih, timeOld)
qb := &fakeQbt{torrents: []qbt.Torrent{{Hash: ih, State: "downloading", AddedOn: addedRecent}}}
w := newTestWorker(st, qb)
if err := w.Poll(context.Background()); err != nil {
t.Fatalf("Poll: %v", err)
}
if got := st.downloads["1"].State; got != store.StateFailed {
t.Errorf("state = %q, want failed (source_gone сверкой не воскрешается)", got)
}
}
// Retry задачи source_gone сбрасывает source_miss_count, чтобы вернувшаяся в
// downloading задача получила полное грейс-окно, а не упала снова на ближайшем
// тике, если переотданная раздача ещё не видна в qBittorrent.
func TestRetryResetsSourceMissCount(t *testing.T) {
const ih = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
st := oneFailed(store.StateFailed, errCodeSourceGone, ih, timeRecent)
st.downloads["1"].SourceMissCount = 3 // задача упала по порогу
qb := &fakeQbt{torrents: nil} // раздачи всё ещё нет (Add вслепую)
w := newTestWorker(st, qb)
w.cfg.SourceMissingThreshold = 3
if err := w.Retry(context.Background(), "1"); err != nil {
t.Fatalf("Retry: %v", err)
}
if got := st.downloads["1"].SourceMissCount; got != 0 {
t.Fatalf("после retry miss = %d, want 0 (сброс)", got)
}
if len(qb.added) != 1 {
t.Errorf("retry без живой раздачи должен переотдать источник, got %d Add", len(qb.added))
}
// Грейс-окно: раздача ещё не видна — задача НЕ падает на ближайшем тике
// (miss стал бы 1 < 3, а не 4 при несброшенном счётчике).
if err := w.Poll(context.Background()); err != nil {
t.Fatalf("Poll after retry: %v", err)
}
if got := st.downloads["1"].State; got != store.StateDownloading {
t.Errorf("state = %q, want downloading (грейс-окно не съедено)", got)
}
if got := st.downloads["1"].SourceMissCount; got != 1 {
t.Errorf("miss = %d, want 1 (один промах после сброса)", got)
}
}
internal/worker/worker.go
+31 -2
View File
@@ -133,6 +133,12 @@ const (
// catch_timeout (устойчивая недоступность qBit). Раздачи в qBittorrent нет,
// восстановлению сверкой не подлежит.
errCodeQbitAdd = "qbit_add"
// errCodeSourceGone — раздача активной (downloading) загрузки устойчиво (после
// дебаунса source_missing_threshold) пропала из qBittorrent: пользователь/другой
// клиент её удалил. Distinct-код, отличный от qbit_error (реальная ошибка qBit) и
// magnet_timeout/stalled (наша нетерпеливость). Восстановлению сверкой НЕ подлежит
// (удаление намеренно) — но задача штатно retriable: Retry заново отдаёт источник.
errCodeSourceGone = "source_gone"
)
// Notifier — исходящие пинги (Telegram). Вызывается неблокирующе.
@@ -501,9 +507,22 @@ func (w *Worker) Poll(ctx context.Context) error {
continue // нечем сопоставить (в Ф1 не случается: magnet всегда с infohash)
}
t, ok := torrentFor(d, byHash)
// Дебаунс пропажи источника у активной загрузки (тот же счётчик, что и
// сверка рассинхрона): промах наращивает source_miss_count, появление
// раздачи сбрасывает его. Устойчивая пропажа (после порога) уводит задачу
// в failed(source_gone) — иначе downloading без раздачи в qBittorrent
// оставался бы вечным зомби (MAJOR-3).
if present := w.debounceSource(ctx, d, ok); !present {
lctx := w.scoped(ctx, capIngest, d.ID, d.PrimaryInfohash())
logctx.From(lctx).Warn("active download source gone from qbittorrent",
"miss_count", d.SourceMissCount+1)
w.transition(lctx, d, store.StateFailed, errCodeSourceGone,
"источник удалён из qBittorrent")
continue
}
if !ok {
w.log.Warn("active download not found in qbittorrent",
"capability", capIngest, "download_id", d.ID, "infohash", d.PrimaryInfohash())
// До порога: транзиентный промах (например рестарт демона qBit) — ждём
// следующий тик, задачу не трогаем.
continue
}
w.captureInfohashes(ctx, d, t)
@@ -839,6 +858,16 @@ func (w *Worker) Retry(ctx context.Context, id string) error {
w.log.Warn("retry basis reset failed",
"capability", capReview, "download_id", id, "error", err)
}
// Сброс счётчика пропусков источника: retried source_gone-задача иначе вошла бы
// в downloading с source_miss_count == threshold и упала бы снова на ближайшем
// тике, если переотданная раздача ещё не видна в выдаче qBittorrent — без
// обещанного грейс-окна (MAJOR-3). Best-effort: сбой лишь лишает свежего окна.
if d.SourceMissCount != 0 {
if err := w.store.SetSourceMissCount(ctx, id, 0); err != nil {
w.log.Warn("retry miss count reset failed",
"capability", capReview, "download_id", id, "error", err)
}
}
logctx.From(w.scoped(ctx, capReview, id, d.PrimaryInfohash())).Info("download retried", "from", d.State)
return nil
}
internal/worker/worker_test.go
+6 -2
View File
@@ -370,15 +370,19 @@ func TestPollMatchesByInfohashV2(t *testing.T) {
}
}
func TestPollIgnoresMissingTorrent(t *testing.T) {
// Единичный транзиентный промах (торрента нет в этом тике) при пороге дебаунса
// > 1 не двигает состояние — устойчивую пропажу и переход в source_gone проверяют
// TestSourceGone* в recovery_test.
func TestPollToleratesTransientMissingTorrent(t *testing.T) {
st := oneDownloading("541adcff3b6dd5dba7088ea83317d9d6fac331d6", timeRecent)
qb := &fakeQbt{torrents: nil} // торрента в qBittorrent нет
w := newTestWorker(st, qb)
w.cfg.SourceMissingThreshold = 3 // порог не достигнут за один тик
if err := w.Poll(context.Background()); err != nil {
t.Fatal(err)
}
if st.downloads["1"].State != store.StateDownloading {
t.Errorf("без торрента состояние не должно меняться, got %q", st.downloads["1"].State)
t.Errorf("единичный промах не должен менять состояние, got %q", st.downloads["1"].State)
}
}